package org.pqw.bookmodel.config;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.pqw.bookmodel.config.enums.MqTagEnum;
import org.pqw.bookmodel.config.enums.MqTopicEnum;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author 白衣雪
 * @since 2021/6/26 13:11
 */
@Slf4j
@Component
public class MqProducer {

    /** 防止多线程下数据不一致*/
    private static AtomicInteger flag = new AtomicInteger(0);

    @Resource
    private DefaultMQProducer defaultProducer;

    public void sendMq(MqTopicEnum topic, MqTagEnum tag,AbstractMqContent content){
        log.info("mq topic:{} tag:{}", topic, tag);
        Message message = new Message();
        message.setTopic(topic.name());
        message.setTags(tag.name());
        message.setBody(content.getContent().getBytes());
        this.start();
        this.doSend(message);
    }

    private void start(){
        try {
            if (flag.get() == 0){
                this.defaultProducer.start();
                flag.getAndIncrement();
            }
            log.info("produce started");
        }catch (Exception e){
            log.info("produce start error");
        }
    }

    private void doSend(Message message){
        try {
            log.info("mq content:{}", JSONUtil.toJsonStr(message.getBody()));
            SendResult result = this.defaultProducer.send(message);
            log.info("mq result:{}", JSONUtil.toJsonStr(result));
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private void doSend(Message message, int failRetryCount){
        try {
            log.info("mq content:{}", JSONUtil.toJsonStr(message.getBody()));
            this.defaultProducer.setRetryTimesWhenSendFailed(failRetryCount);
            SendResult result = this.defaultProducer.send(message);
            log.info("mq result:{}", JSONUtil.toJsonStr(result));
        }catch (Exception e){
            log.info("mq send error");
        }
    }

    /**
     *
     * @param topic 主题
     * @param tag 标签
     * @param content 发送内容
     * @param sendCallback 异步的回调
     * @param failRetryCount 失败重发次数
     */
    public void sendMqSync(MqTopicEnum topic, MqTagEnum tag,AbstractMqContent content, SendCallback sendCallback, int failRetryCount){
        log.info("mq topic:{} tag:{}", topic, tag);
        Message message = new Message();
        message.setTopic(topic.name());
        message.setTags(tag.name());
        message.setBody(content.getContent().getBytes());
        this.start();
        this.doSendSync(message, sendCallback, failRetryCount);
    }

    public void doSendSync(Message message, SendCallback sendCallback, int failRetryCount){
        try {
            this.defaultProducer.setRetryTimesWhenSendAsyncFailed(failRetryCount);
            this.defaultProducer.send(message, sendCallback);
        }catch (Exception e){
            log.info("mq send error");
        }
    }

}
